package com.oct.ywsc.job;

import com.alibaba.fastjson2.JSONObject;
import com.oct.ywsc.core.JsonUtil;
import com.oct.ywsc.core.bean.MsgBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSession;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 消息转换任务
 */
@Slf4j
public class MsgTransTask implements Runnable {

    // 待处理消息
    private  MsgBean record;
    // 所有路由规则
    private RuleManage ruleManage;

    private final String key_table = "table";
    private final String key_op_type = "op_type";


    private JdbcTemplate jdbcTemplate;

    private CustomerKafkaProducer customerKafkaProducer;

    /**
     * 消息转换任务
     * @param record kafka消息
     * @param ruleManage 消息转换规则管理器
     *
     */
    public MsgTransTask(MsgBean record, RuleManage ruleManage
                        ,JdbcTemplate jdbcTemplate
    ,CustomerKafkaProducer customerKafkaProducer) {
        this.record = record;
        this.ruleManage = ruleManage;
        this.jdbcTemplate = jdbcTemplate;
        this.customerKafkaProducer = customerKafkaProducer;
    }

    @Override
    public void run() {
        // 根据规则处理消息
        log.info(String.format("开始处理消息offset[%s],value %s", record.getOffset(), record.getMsg()));
        // 将消息体转换为map
        Map<String,String> msg = recodeToMap(record.getMsg());

        // 根据规则配置进行消息转换
        this.doTrans(msg,record.getOffset());

    }


    /**
     * 转换消息
     * @param msg
     * @param offset 消息的offset
     */
    public void doTrans(Map<String,String> msg,Long offset){



        try {
            for (TransRule transRule : this.ruleManage.ruleList()) {
                if (transRule.isMatch(msg.get(key_table), msg.get(key_op_type))) {

                        transRule.setJdbcTemplate(this.jdbcTemplate);
                        List<Map<String, Object>> result = transRule.doTrans(msg);

                        log.info(String.format("消息offset[%s]匹配规则%s sql执行结果为[%s]", offset, transRule.getName(), JSONObject.toJSONString(result)));

                        if (!CollectionUtils.isEmpty(result) && !CollectionUtils.isEmpty(transRule.sendTopic())) {
                            for (String topic : transRule.sendTopic()) {
                                customerKafkaProducer.send(topic, this.formatSendMsg(result, transRule.sendMsgField()));
                            }
                        }


                } else{
                    log.info(String.format("消息offset[%s]不匹配规则%s,跳过", offset, transRule.getName()));
                }
            }

        } catch (Exception e) {
            log.error(String.format("消息offset:[%s],value[%s]失败",offset,JSONObject.toJSONString(msg)), e);
        }
        log.info(String.format("消息offset[%s]处理完成", offset));
    }






    /**
     * 将kafka消息转换成map对象
     * @param record
     * @return
     */
    private Map<String,String> recodeToMap(String record) {
         JSONObject jo =  JSONObject.parseObject(record);

         return JsonUtil.toMap(jo);
    }

    /**
     * 格式化发送消息
     * @param result
     * @param sendMsgKeyList
     * @return
     */
    private String formatSendMsg(List<Map<String,Object>> result,List<String> sendMsgKeyList){
        String sendMsg = null;
        if (result.size() > 1){
            List<Map<String,String>> sendMsgList = new ArrayList<>();
            for (Map<String,Object> map : result){
                Map<String,String> sendMsgMap = new HashMap<>();
                for (String key : sendMsgKeyList){
                    if (map.containsKey(key.toUpperCase()) ) {
                        sendMsgMap.put(key, map.get(key.toUpperCase()).toString());
                    }
                    if( map.containsKey(key.toLowerCase())){
                        sendMsgMap.put(key, map.get(key.toLowerCase()).toString());
                    }
                }
                sendMsgList.add(sendMsgMap);
            }
            sendMsg = JSONObject.toJSONString(sendMsgList);
        }else{
            Map<String,String> sendMsgMap = new HashMap<>();
            for (String key : sendMsgKeyList){
                if (result.get(0).containsKey(key.toUpperCase()) ) {
                    sendMsgMap.put(key, result.get(0).get(key.toUpperCase()).toString());
                }
                if( result.get(0).containsKey(key.toLowerCase())){
                    sendMsgMap.put(key, result.get(0).get(key.toLowerCase()).toString());
                }
            }
            sendMsg = JSONObject.toJSONString(sendMsgMap);
        }
        return sendMsg;
    }









}
